/***
* ==++==
*
* Copyright (c) Microsoft Corporation.  All rights reserved.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* procinfo.h
*
* C++ Actors: actor information, data structures internal to the library.
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/
#pragma once

#include <memory>

#include "actor_public.h"
#include "actor_serialization.h"
#include "xxagents.h"

//
// This file contains declarations that are almost entirely private to the runtime and
// of little interest to users of the C++ Actors library.
//
namespace actors
{

class tcp_proxy;
class tcp_listener;

enum DummyType { DummyNone };


#pragma region Internal Message Dispatch Helpers

//
// IMessageDispatcher represents all targets of internal messages
//
class _IMessageDispatcher
{
public:
    _IMessageDispatcher() : m_once(false) {}

    virtual bool dispatch(dispatch_msg *msg) = 0;
    bool OnceOnly() const { return m_once; }

    virtual bool is_local() const { return true; }

protected:
    _IMessageDispatcher(bool once) : m_once(once) {}

private:
    bool m_once;
};

typedef std::shared_ptr<_IMessageDispatcher> IMessageDispatcher;

//
// A message dispatcher defined from a functor.
//
template<typename Functor>
class _Dispatcher : public _IMessageDispatcher
{
public:
    _Dispatcher(bool once, Functor functor) : _IMessageDispatcher(once), func(functor) {}
    bool dispatch(dispatch_msg *dmsg) { return func(reinterpret_cast<general_msg *>(dmsg)); }

private:
    _Dispatcher &operator =(const _Dispatcher &);

    Functor func;
};

//
// A message dispatcher defined from a ITarget<T>
//
template<typename Payload>
class _Dispatcher2 : public _IMessageDispatcher
{
public:
    _Dispatcher2(bool once, Concurrency::ITarget<Payload> *target) : _IMessageDispatcher(once), target(target) {}

    bool dispatch(dispatch_msg *dmsg)
    {
        if ( dmsg == nullptr ) return false;

        bool result = false;

        general_msg *msg = reinterpret_cast<general_msg *>(dmsg);

        Payload data;

        bool same_type = msg->Matches(typeid(Payload).name());

        if ( same_type && _actor_info::process_message(msg, data) )
        {
            result = Concurrency::asend(target, data);
        }

        if ( result ) delete dmsg;

        return result;
    }
private:
    Concurrency::ITarget<Payload> *target;
};

template<typename Functor>
class _Continuation : public _IMessageDispatcher
{
public:
    _Continuation(Functor functor) : _IMessageDispatcher(false), func(functor) {}
    _Continuation & operator =(const _Continuation &other) { this->func = other.func; return *this; }
    bool dispatch(dispatch_msg *dmsg) { func(); return true;}
private:
    Functor func;
};

template<typename Functor>
IMessageDispatcher CreateForwarder(bool once, Functor functor)
{
    return IMessageDispatcher(new _Dispatcher<Functor>(once, functor));
}

template<typename Functor>
IMessageDispatcher CreateContinuation(Functor functor)
{
    return IMessageDispatcher(new _Continuation<Functor>(functor));
}

template<typename Payload>
IMessageDispatcher CreateForwarder(bool once, Concurrency::ITarget<Payload> *target)
{
    return IMessageDispatcher(new _Dispatcher2<Payload>(once, target));
}

#pragma endregion

#pragma region Internal Serialization Help

class _msg_serializer : public Serializer
{
#define _msg_serializer_BLOCK_SIZE 4096

public:

    _msg_serializer(compression *compress, Serializer &serializer, bool trace);
    ~_msg_serializer();

    bool start(msg_op cmd, msg_guid &guid, const std::string &type_name);

    /// <summary>
    /// Write a number of bytes to the underlying serialization target and advance the internal write position.
    /// Once 'BlockSize' bytes have been buffered, write to the underlying serializer.
    /// </summary>
    /// <remarks>
    /// Writing zero bytes always succeeds, whether the underlying buffer, stream, socket, etc. is valid or not
    /// </remarks>
    virtual bool write(size_t byteCount, const void *source);

    /// <summary>
    /// Finalize a top-level serialization request (message send).
    /// </summary>
    virtual bool flush();

private:
    _msg_serializer &operator=(const _msg_serializer &);

    void _init_header(size_t size);

    compression *m_compress;
    Serializer  &m_serializer;
    size_t       m_offset;
    bool         m_trace;

    union {
        block_header  m_header;
        byte          m_block[_msg_serializer_BLOCK_SIZE];
    };
#undef _BLOCK_SIZE
};
#pragma endregion

//
// actor_info, an internal class used to hide the details of PIDs
// 
class _actor_info
{
public:

    _DISPIMP _actor_info () : faulted(false), chan(nullptr), process(nullptr) { memset(&this->guid, 0, sizeof(GUID)); }
    _DISPIMP _actor_info (const msg_guid &id, const std::string &uri);

    _DISPIMP operator std::string() { return to_string(); }
    _DISPIMP ~_actor_info () { chan.reset(); process = nullptr; }

    template<typename Payload>
    static bool process_message(general_msg *msg, Payload &data)
    {
        if ( msg == nullptr ) return false;

        switch(msg->cmd)
        {
        case CMD_MESSAGE:

            if ( msg->msg_length >= 0 )
            {
                msg_list_serializer serializer(msg);

                return ::deserialize(data, serializer);
            }
            else
            {
                data = reinterpret_cast<local_msg<Payload> *>(msg)->data;
                return true;
            }
        }

        return false;
    }

#pragma region "private declarations"
private:
    _DISPIMP friend bool (::deserialize)(actor_info p_info, actors::Serializer &serializer);

    _DISPIMP static Serializer *_get_msg_serializer(msg_op cmd, const std::string &type_name, actor_info p_info, bool trace);

    template<typename T>
    local_msg<T> * _allocate_local(const type_info &info, bool trace)
    {
        local_msg<T> *msg = new local_msg<T>();
        msg->flags = trace ? MSGFLG_TRACE : MSGFLG_NONE;
        msg->cmd = CMD_MESSAGE;
        msg->guid = this->guid;
        msg->msg_length = -1;
        msg->name = info.name();
        return msg;
    }

    _DISPIMP std::string to_string();

    _DISPIMP bool is_local() const;

    _DISPIMP _actor_info (tcp_listener *);
    
    _DISPIMP _actor_info (const std::string &uri) : uri(uri), chan(nullptr), faulted(false), process(nullptr) { }
    _DISPIMP _actor_info (const std::string &uri, const msg_guid &guid) : uri(uri), chan(nullptr), faulted(false), process(nullptr), guid(guid) { }

    _DISPIMP static actor_info _spawn_remote(const std::string &resource, const std::string &uri);
    _DISPIMP static actor_info _spawn_link_remote(const std::string &resource, PID link, const std::string &uri);
    _DISPIMP static actor_info _spawn_remote(const std::string &resource, actor_info);
    _DISPIMP static actor_info _respawn_remote(const std::string &resource, actor_info, const std::vector<byte> &);
    _DISPIMP static bool _spawn_remote(msg_op cmd, const std::string &resource, actor_info, IMessageDispatcher proxy, PID link);

    _DISPIMP static IMessageDispatcher _spawn(tcp_listener *, PID link, const msg_guid &guid, const char *type_name);
    _DISPIMP static IMessageDispatcher _spawn(tcp_listener *, const msg_guid &guid, const char *type_name);
    _DISPIMP static IMessageDispatcher _respawn(tcp_listener *, general_msg *msg);

    _DISPIMP static actor_info _recover(tcp_listener *, const msg_guid &guid);

    friend class PID;
    friend class actor;
    friend class tcp_listener;
    friend class tcp_proxy;
    friend struct dispatch_msg;
    friend class _Socket_msg_serializer;

    _DISPIMP friend bool (::deserialize)(PID &pid, actors::Serializer &serializer);
    _DISPIMP friend bool (::serialize)(const PID pid, actors::Serializer &serializer);
    _DISPIMP friend size_t (::calculate_size)(const PID pid);

    msg_guid guid;
    std::string uri;
    IMessageDispatcher chan;
    actor *process;

    bool faulted;
#pragma endregion
};

} // namespace actors
